package com.zhiwen.util;

import com.zhiwen.exception.LockException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @Auther: wuzhiwen
 * @Date: 2018/11/11 15:47
 * @Description:
 */
public class DistributedLock implements Lock, Watcher {
    //zookeeper连接
    private ZooKeeper zooKeeper = null;

    //会话时间
    private int sessionTimeout = 30000;

    //根节点
    private String rootLock = "/rootlock";

    //计数器
    private CountDownLatch countDownLatch;

    //异常集合
    private List<Exception> exceptionList = new ArrayList<Exception>();

    //等待前一个锁
    private String WAIT_LOCK;

    //竞争的资源
    private String lockName;

    //当前锁
    // 当前锁
    private String CURRENT_LOCK;

    public DistributedLock(String config, String lockName) {
        this.lockName = lockName;
        try {
            zooKeeper = new ZooKeeper(config, sessionTimeout, this);
            //判断根节点是否存在, false则不是有this.watchManager.defaultWatcher 默认的Watcher
            Stat stat = zooKeeper.exists(rootLock, false);
            if (stat == null) {
                //path：创建节点路径，需保证父节点已存在 ,data：节点数据 acl:权限列表, createMode:节点类型
                /*
                 *  acl:权限列表
                 *      OPEN_ACL_UNSAFE：完全开放
                 *      CREATOR_ALL_ACL：创建该znode的连接拥有所有权限
                 *      READ_ACL_UNSAFE：所有的客户端都可读
                 *
                 *   createMode:节点类型
                 *      PERSISTENT：持久化节点
                 *      PERSISTENT_SEQUENTIAL：持久化有序节点
                 *      EPHEMERAL：临时节点（连接断开自动删除）
                 *      EPHEMERAL_SEQUENTIAL：临时有序节点（连接断开自动删除）
                 *
                 */
                zooKeeper.create(rootLock, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }

        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void lock() {
        //对异常进行判断
        if (exceptionList.size() > 0) {
            throw new LockException(exceptionList.get(0));
        }
        //获取锁
        if (this.tryLock()) {
            System.out.println(Thread.currentThread().getName() + " " + lockName + "获得了锁");
            return;
        } else {
            waitLocak(WAIT_LOCK, sessionTimeout);
        }
    }

    //设置监听
    private boolean waitLocak(String wait_lock, long sessionTimeout) {
        try {
            Stat stat = zooKeeper.exists(rootLock + "/" + wait_lock, true);
            if (stat != null) {
                //创建计数
                this.countDownLatch = new CountDownLatch(1);
                // 计数等待，若等到前一个节点消失，则precess中进行countDown，停止等待，获取锁
                this.countDownLatch.await(sessionTimeout, TimeUnit.MILLISECONDS);
                this.countDownLatch = null;
            }
            return true;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        this.lock();
    }

    @Override
    public boolean tryLock() {
        try {
            //设置子节点命名规则，便于后别切割分别和排序
            String lockSplit = "_lock_";
            String sonName = rootLock + "/" + lockName + lockSplit;
            //创建临时子节点
            CURRENT_LOCK = zooKeeper.create(sonName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            //获取所有子节点列表
            List<String> children = zooKeeper.getChildren(rootLock, false);

            List<String> sonList = new ArrayList<>();
            //判断是否是最小节点
            for (String child : children) {
                //将所有节点根据命名规则切割
                String[] split = child.split(lockSplit);
                //只将该资源下的节点保存
                if (split[0].equals(lockName)) {
                    sonList.add(child);
                }
                //排序
                Collections.sort(sonList);
                //判断最小
                if (CURRENT_LOCK.equals(rootLock + "/" + sonList.get(0))) {
                    return true;
                }
                //这样的写法相对的是遍历查询，效率不是很高，
                //如果不是 则找到自己前面的节点
                for (int i = 0; i < sonList.size(); i++) {
                    if (CURRENT_LOCK.equals(rootLock + "/" + sonList.get(i))) {
                        //对前一个小的节点进行定位
                        WAIT_LOCK = sonList.get(--i);
                        //在这里设置监听会好点 我觉得
                        break;
                    }
                }
                // 若不是最小节点，则找到自己的前一个节点
                // 因为我们已经进行了排序，所以使用集合自带Collections.binarySearch()的方法，二分查找效率会快点。
//                String prevNode = CURRENT_LOCK.substring(CURRENT_LOCK.lastIndexOf("/") + 1);
//                WAIT_LOCK = sonList.get(Collections.binarySearch(sonList, prevNode) - 1);
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        if (this.tryLock()) {
            return true;
        }
        return waitLocak(WAIT_LOCK, time);
    }

    @Override
    public void unlock() {
        try {
            System.out.println("删除锁");
            zooKeeper.delete(CURRENT_LOCK, -1);
            CURRENT_LOCK = null;
            zooKeeper.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    /**
     * 节点监视器
     *
     * @param watchedEvent
     */
    @Override
    public void process(WatchedEvent watchedEvent) {
        if (countDownLatch != null) {
            this.countDownLatch.countDown();
        }
    }

    /**
     * 优化lock
     *
     * @return
     */
    public boolean optimizeTryLock() {

        //设置一个标记
        boolean isDown = false;
        Stat stat = null;
        //后期如果报了异常可以对他进行重试
        while (!isDown) {
            try {
                isDown = true;

                //设置子节点命名规则，便于后别切割分别和排序
                String lockSplit = "_lock_";
                String sonName = rootLock + "/" + lockName + lockSplit;
                stat = zooKeeper.exists(CURRENT_LOCK, false);
                if (stat == null) {
                    //创建临时子节点
                    CURRENT_LOCK = zooKeeper.create(sonName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
                }
                //获取所有子节点列表
                List<String> children = zooKeeper.getChildren(rootLock, false);
                List<String> sonList = new ArrayList<>();
                //判断是否是最小节点
                for (String child : children) {
                    //将所有节点根据命名规则切割
                    String[] split = child.split(lockSplit);
                    //只将该资源下的节点保存
                    if (split[0].equals(lockName)) {
                        sonList.add(child);
                    }
                    //排序
                    Collections.sort(sonList);
                    //判断最小
                    if (CURRENT_LOCK.equals(rootLock + "/" + sonList.get(0))) {
                        return true;
                    }
                    // 若不是最小节点，则找到自己的前一个节点
                    // 因为我们已经进行了排序，所以使用集合自带Collections.binarySearch()的方法，二分查找效率会快点。
                    String prevNode = CURRENT_LOCK.substring(CURRENT_LOCK.lastIndexOf("/") + 1);
                    WAIT_LOCK = sonList.get(Collections.binarySearch(sonList, prevNode) - 1);
                }
                return waitLocak(WAIT_LOCK, sessionTimeout, stat);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            isDown = false;
        }
        return false;
    }

    //设置监听
    private boolean waitLocak(String wait_lock, long sessionTimeout, Stat stat) {
        try {
            zooKeeper.getData(rootLock + "/" + wait_lock, true, stat);
            if (stat != null) {
                //创建计数
                this.countDownLatch = new CountDownLatch(1);
                // 计数等待，若等到前一个节点消失，则precess中进行countDown，停止等待，获取锁
                this.countDownLatch.await(sessionTimeout, TimeUnit.MILLISECONDS);
                this.countDownLatch = null;
            }
            return true;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }


}
